package com.atguigu.gmall.realtime.utils;

import com.alibaba.fastjson.JSON;
import com.atguigu.gmall.realtime.app.marketing.bean.GmallEvent;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.List;
import java.util.Properties;

public class MyKafkaUtil {

    static String BOOTSTRAP_SERVERS = "hadoop102:9092, hadoop103:9092, hadoop104:9092";
    static String DEFAULT_TOPIC = "default_topic";

    public static FlinkKafkaConsumer<String> getKafkaConsumer(String topic, String groupId) {
        Properties prop = new Properties();
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        prop.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed");

        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(topic,
            new KafkaDeserializationSchema<String>() {
                @Override
                public boolean isEndOfStream(String nextElement) {
                    return false;
                }

                @Override
                public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                    if (record != null && record.value() != null) {
                        return new String(record.value());
                    }
                    return null;
                }

                @Override
                public TypeInformation<String> getProducedType() {
                    return TypeInformation.of(String.class);
                }
            }, prop);
        return consumer;
    }

    public static FlinkKafkaConsumer<GmallEvent> getFlinkKafkaConsumer(List<String> topics, String groupId) {

        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        return new FlinkKafkaConsumer<GmallEvent>(topics,
                new KafkaDeserializationSchema<GmallEvent>() {
                    @Override
                    public TypeInformation<GmallEvent> getProducedType() {
                        return TypeInformation.of(GmallEvent.class);
                    }

                    @Override
                    public boolean isEndOfStream(GmallEvent nextElement) {
                        return false;
                    }

                    @Override
                    public GmallEvent deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                        if (record == null || record.value() == null) {
                            return null;
                        }
                        return  new GmallEvent( record.topic(), JSON.parseObject(new String(record.value())));
                    }
                },
                properties);
    }



    public static FlinkKafkaProducer<String> getKafkaProducer(String topic) {

        Properties prop = new Properties();
        prop.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        prop.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 60 * 15 * 1000 + "");

        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<String>(DEFAULT_TOPIC, new KafkaSerializationSchema<String>() {

            @Override
            public ProducerRecord<byte[], byte[]> serialize(String jsonStr, @Nullable Long timestamp) {
                return new ProducerRecord<byte[], byte[]>(topic, jsonStr.getBytes());
            }
        }, prop, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

        return producer;
    }

    //获取生产者对象（由调用者提供序列化）
    public static <T>FlinkKafkaProducer<T> getKafkaProducerBySchema(KafkaSerializationSchema kafkaSerializationSchema) {

        Properties prop = new Properties();
        prop.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        prop.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 60 * 15 * 1000 + "");

        FlinkKafkaProducer<T> producer = new FlinkKafkaProducer<T>(DEFAULT_TOPIC, kafkaSerializationSchema , prop, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

        return producer;
    }

    //获取从topic_db主题中读取数据 创建动态表的建表语句
    public static String getTopicDbDDL(String groupId) {
        return "CREATE TABLE topic_db (\n" +
            "    `database` string,\n" +
            "    `table` string,\n" +
            "    `type` string,\n" +
            "    `ts` string,\n" +
            "    `data` MAP<string, string>,\n" +
            "    `old` MAP<string, string>,\n" +
            "    proc_time as proctime()\n" +
            ") " + getKafkaDDL("topic_db", groupId);
    }

    //获取kafka连接器连接属性
    public static String getKafkaDDL(String topic, String groupId) {
        return "WITH (\n" +
            "  'connector' = 'kafka',\n" +
            "  'topic' = '" + topic + "',\n" +
            "  'properties.bootstrap.servers' = '" + BOOTSTRAP_SERVERS + "',\n" +
            "  'properties.group.id' = '" + groupId + "',\n" +
            "  'scan.startup.mode' = 'group-offsets',\n" +
            "  'format' = 'json'\n" +
            ")";
    }

    //获取upsert-kafak连接器连接属性
    public static String getUpsertKafkaDDL(String topic) {
        return " WITH (\n" +
            "  'connector' = 'upsert-kafka',\n" +
            "  'topic' = '" + topic + "',\n" +
            "  'properties.bootstrap.servers' = '" + BOOTSTRAP_SERVERS + "',\n" +
            "  'key.format' = 'json',\n" +
            "  'value.format' = 'json'\n" +
            ")";
    }

}
